package chapter8

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object KafkaStreaming extends App{

  val conf: SparkConf = new SparkConf().setAppName(this.getClass.getName).setMaster("local[*]")

  val sc = new SparkContext(conf)

  val ssc = new StreamingContext(sc, Seconds(5))

  val num = new LongAccumulator()

  sc.register(num)

  val kafkaParams: Map[String, Object] = Map[String, Object](
//    "bootstrap.servers" -> "localhost:9092,anotherhost:9092",
    "bootstrap.servers" -> "master:9092",
    "key.deserializer" -> classOf[StringDeserializer],
    "value.deserializer" -> classOf[StringDeserializer],
    "group.id" -> "use_a_separate_group_id_for_each_stream",
    "auto.offset.reset" -> "latest",
    "enable.auto.commit" -> (false: java.lang.Boolean),
    "receive.buffer.bytes" -> (65536: java.lang.Integer)
  )

  val topics: Array[String] = Array("topicA", "topicB")
  val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
    ssc,
    PreferConsistent,
    Subscribe[String, String](topics, kafkaParams)
  )

  stream.map(record => (record.key, record.value)).print()

  ssc.start()
  ssc.awaitTermination()

}
